package com.taotao.slink.utils.localstore;

import com.alibaba.fastjson.JSON;
import org.rocksdb.Options;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.rocksdb.RocksIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 本地存储容器,用来替代 WAL 功能
 */
public class LocalStore {

    private final static Logger logger = LoggerFactory.getLogger(LocalStore.class);


    private final Serializer serializer;
    private final String PLATFORM = "short_link";

    private final String module;

    private final String baseDir;

    private volatile boolean inited = false;

    static {
        RocksDB.loadLibrary();
    }

    private RocksDB rocksDB;


    /**
     * @param basePath 结尾不带 路径分隔符
     * @param module
     */
    public LocalStore(String basePath, String module, Serializer serializer) {
        this.module = module;
        this.serializer = serializer;
        baseDir = basePath + File.separator + PLATFORM + File.separator + this.module;
        ensureDirOK(baseDir);
        logger.info("recovery目录:" + baseDir);
    }

    public LocalStore(String basePath, String module) {
        this(basePath, module, new HessianSerializer());
    }


    public LocalStore(String module) {
        this(System.getProperty("user.home"), module, new HessianSerializer());
    }

    public LocalStore(String module, Serializer serializer) {
        this(System.getProperty("user.home"), module, serializer);
    }

    public synchronized void start() throws Exception {
        if (this.inited) {
            return;
        }
        Options options = new Options();
        options.setCreateIfMissing(true);
        rocksDB = RocksDB.open(options, baseDir);
        this.inited = true;
    }

    public <T> void put(String key, T data) throws Exception {
        checkStart();
        put(key, serializer.serialize(data));
    }

    public <T> T get(String key, Class<T> clazz) throws UnsupportedEncodingException, RocksDBException {
        byte[] value = getRawValue(key);
        if (value == null) {
            return null;
        }
        return JSON.parseObject(value, clazz);
    }

    public void put(String key, byte[] value) throws RocksDBException, UnsupportedEncodingException {
        checkKey(key);
        checkStart();
        rocksDB.put(encodeKey(key), value);
    }

    public byte[] getRawValue(String key) throws UnsupportedEncodingException, RocksDBException {
        checkKey(key);
        checkStart();
        byte[] value = this.rocksDB.get(encodeKey(key));
        return value;
    }

    public void delete(String key) throws UnsupportedEncodingException, RocksDBException {
        checkKey(key);
        checkStart();
        this.rocksDB.delete(encodeKey(key));
    }

    public <T> List<T> queryList(int maxCount, Class<T> clazz) throws Exception {
        checkStart();
        List<T> dataList = new ArrayList<>(maxCount);
        RocksIterator iterator = this.rocksDB.newIterator();
        iterator.seekToFirst();
        while (iterator.isValid()) {
            byte[] value = iterator.value();
            T data = serializer.deserialize(value, clazz);
            dataList.add(data);
            iterator.next();
        }
        return dataList;
    }


    public <T> Map<String, T> queryMap(int maxCount, Class<T> clazz) throws Exception {
        checkStart();
        Map<String, T> dataList = new HashMap<>(maxCount);
        RocksIterator iterator = this.rocksDB.newIterator();
        iterator.seekToFirst();
        while (iterator.isValid()) {
            T data = serializer.deserialize(iterator.value(), clazz);
            dataList.put(decodeKey(iterator.key()), data);
            iterator.next();
        }
        return dataList;
    }


    private void checkKey(String key) {
        if (key == null) {
            throw new IllegalArgumentException("key 不能为空");
        }
    }


    private void checkStart() {
        if (!inited) {
            throw new IllegalStateException("LocalStore未初始化,使用前先调用start()");
        }
    }

    private byte[] encodeKey(String key) throws UnsupportedEncodingException {
        checkKey(key);
        return key.getBytes("UTF-8");
    }

    private String decodeKey(byte[] key) throws UnsupportedEncodingException {
        return new String(key, "UTF-8");
    }

    public void shutdown() {
        if (this.rocksDB != null) {
            this.rocksDB.close();
        }
    }

    public static void ensureDirOK(final String dirName) {
        if (dirName != null) {
            File f = new File(dirName);
            if (!f.exists()) {
                boolean result = f.mkdirs();
                logger.info(dirName + " mkdir " + (result ? "OK" : "Failed"));
            }
        }
    }
    
}
